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.40 by root, Tue May 27 05:36:27 2008 UTC vs.
Revision 1.84 by root, Thu Aug 21 19:13:05 2008 UTC

1package AnyEvent::Handle; 1package AnyEvent::Handle;
2 2
3no warnings; 3no warnings;
4use strict; 4use strict qw(subs vars);
5 5
6use AnyEvent (); 6use AnyEvent ();
7use AnyEvent::Util qw(WSAWOULDBLOCK); 7use AnyEvent::Util qw(WSAEWOULDBLOCK);
8use Scalar::Util (); 8use Scalar::Util ();
9use Carp (); 9use Carp ();
10use Fcntl (); 10use Fcntl ();
11use Errno qw/EAGAIN EINTR/; 11use Errno qw(EAGAIN EINTR);
12 12
13=head1 NAME 13=head1 NAME
14 14
15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent 15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
16 16
17=cut 17=cut
18 18
19our $VERSION = '0.04'; 19our $VERSION = 4.232;
20 20
21=head1 SYNOPSIS 21=head1 SYNOPSIS
22 22
23 use AnyEvent; 23 use AnyEvent;
24 use AnyEvent::Handle; 24 use AnyEvent::Handle;
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> or C<EBADMSG>). 118error (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 119
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.
151
152=item timeout => $fractional_seconds
153
154If non-zero, then this enables an "inactivity" timeout: whenever this many
155seconds pass without a successful read or write on the underlying file
156handle, the C<on_timeout> callback will be invoked (and if that one is
157missing, an C<ETIMEDOUT> error will be raised).
158
159Note that timeout processing is also active when you currently do not have
160any outstanding read or write requests: If you plan to keep the connection
161idle then you should disable the timout temporarily or ignore the timeout
162in the C<on_timeout> callback.
163
164Zero (the default) disables this timeout.
165
166=item on_timeout => $cb->($handle)
167
168Called whenever the inactivity timeout passes. If you return from this
169callback, then the timeout will be reset as if some activity had happened,
170so this condition is not fatal in any way.
124 171
125=item rbuf_max => <bytes> 172=item rbuf_max => <bytes>
126 173
127If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>) 174If 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 175when 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 179be 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 180(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 181amount of data without a callback ever being called as long as the line
135isn't finished). 182isn't finished).
136 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
137=item read_size => <bytes> 208=item read_size => <bytes>
138 209
139The 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
140on each [loop iteration). Default: C<4096>. 211during each (loop iteration). Default: C<8192>.
141 212
142=item low_water_mark => <bytes> 213=item low_water_mark => <bytes>
143 214
144Sets 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
145buffer: 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
146considered empty. 217considered empty.
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.
147 229
148=item tls => "accept" | "connect" | Net::SSLeay::SSL object 230=item tls => "accept" | "connect" | Net::SSLeay::SSL object
149 231
150When this parameter is given, it enables TLS (SSL) mode, that means it 232When this parameter is given, it enables TLS (SSL) mode, that means it
151will start making tls handshake and will transparently encrypt/decrypt 233will start making tls handshake and will transparently encrypt/decrypt
160You can also provide your own TLS connection object, but you have 242You can also provide your own TLS connection object, but you have
161to make sure that you call either C<Net::SSLeay::set_connect_state> 243to 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 244or C<Net::SSLeay::set_accept_state> on it before you pass it to
163AnyEvent::Handle. 245AnyEvent::Handle.
164 246
165See the C<starttls> method if you need to start TLs negotiation later. 247See the C<starttls> method if you need to start TLS negotiation later.
166 248
167=item tls_ctx => $ssl_ctx 249=item tls_ctx => $ssl_ctx
168 250
169Use the given Net::SSLeay::CTX object to create the new TLS connection 251Use the given Net::SSLeay::CTX object to create the new TLS connection
170(unless a connection object was specified directly). If this parameter is 252(unless a connection object was specified directly). If this parameter is
172 254
173=item json => JSON or JSON::XS object 255=item json => JSON or JSON::XS object
174 256
175This is the json coder object used by the C<json> read and write types. 257This is the json coder object used by the C<json> read and write types.
176 258
177If you don't supply it, then AnyEvent::Handle will use C<encode_json> and 259If you don't supply it, then AnyEvent::Handle will create and use a
178C<decode_json>. 260suitable one, which will write and expect UTF-8 encoded JSON texts.
179 261
180Note that you are responsible to depend on the JSON module if you want to 262Note that you are responsible to depend on the JSON module if you want to
181use this functionality, as AnyEvent does not have a dependency itself. 263use this functionality, as AnyEvent does not have a dependency itself.
182 264
183=item filter_r => $cb 265=item filter_r => $cb
202 if ($self->{tls}) { 284 if ($self->{tls}) {
203 require Net::SSLeay; 285 require Net::SSLeay;
204 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 286 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
205 } 287 }
206 288
207 $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; 289 $self->{_activity} = AnyEvent->now;
208 $self->on_error (delete $self->{on_error}) if $self->{on_error}; 290 $self->_timeout;
291
209 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 292 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain};
210 $self->on_read (delete $self->{on_read} ) if $self->{on_read}; 293 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay};
211 294
212 $self->start_read; 295 $self->start_read
296 if $self->{on_read};
213 297
214 $self 298 $self
215} 299}
216 300
217sub _shutdown { 301sub _shutdown {
218 my ($self) = @_; 302 my ($self) = @_;
219 303
304 delete $self->{_tw};
220 delete $self->{_rw}; 305 delete $self->{_rw};
221 delete $self->{_ww}; 306 delete $self->{_ww};
222 delete $self->{fh}; 307 delete $self->{fh};
223}
224 308
309 $self->stoptls;
310
311 delete $self->{on_read};
312 delete $self->{_queue};
313}
314
225sub error { 315sub _error {
226 my ($self) = @_; 316 my ($self, $errno, $fatal) = @_;
227 317
228 {
229 local $!;
230 $self->_shutdown; 318 $self->_shutdown
231 } 319 if $fatal;
232 320
233 $self->{on_error}($self) 321 $! = $errno;
322
234 if $self->{on_error}; 323 if ($self->{on_error}) {
235 324 $self->{on_error}($self, $fatal);
325 } else {
236 Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; 326 Carp::croak "AnyEvent::Handle uncaught error: $!";
327 }
237} 328}
238 329
239=item $fh = $handle->fh 330=item $fh = $handle->fh
240 331
241This method returns the file handle of the L<AnyEvent::Handle> object. 332This method returns the file handle of the L<AnyEvent::Handle> object.
260 351
261=cut 352=cut
262 353
263sub on_eof { 354sub on_eof {
264 $_[0]{on_eof} = $_[1]; 355 $_[0]{on_eof} = $_[1];
356}
357
358=item $handle->on_timeout ($cb)
359
360Replace the current C<on_timeout> callback, or disables the callback
361(but not the timeout) if C<$cb> = C<undef>. See C<timeout> constructor
362argument.
363
364=cut
365
366sub on_timeout {
367 $_[0]{on_timeout} = $_[1];
368}
369
370=item $handle->autocork ($boolean)
371
372Enables or disables the current autocork behaviour (see C<autocork>
373constructor argument).
374
375=cut
376
377=item $handle->no_delay ($boolean)
378
379Enables or disables the C<no_delay> setting (see constructor argument of
380the same name for details).
381
382=cut
383
384sub no_delay {
385 $_[0]{no_delay} = $_[1];
386
387 eval {
388 local $SIG{__DIE__};
389 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1];
390 };
391}
392
393#############################################################################
394
395=item $handle->timeout ($seconds)
396
397Configures (or disables) the inactivity timeout.
398
399=cut
400
401sub timeout {
402 my ($self, $timeout) = @_;
403
404 $self->{timeout} = $timeout;
405 $self->_timeout;
406}
407
408# reset the timeout watcher, as neccessary
409# also check for time-outs
410sub _timeout {
411 my ($self) = @_;
412
413 if ($self->{timeout}) {
414 my $NOW = AnyEvent->now;
415
416 # when would the timeout trigger?
417 my $after = $self->{_activity} + $self->{timeout} - $NOW;
418
419 # now or in the past already?
420 if ($after <= 0) {
421 $self->{_activity} = $NOW;
422
423 if ($self->{on_timeout}) {
424 $self->{on_timeout}($self);
425 } else {
426 $self->_error (&Errno::ETIMEDOUT);
427 }
428
429 # callback could have changed timeout value, optimise
430 return unless $self->{timeout};
431
432 # calculate new after
433 $after = $self->{timeout};
434 }
435
436 Scalar::Util::weaken $self;
437 return unless $self; # ->error could have destroyed $self
438
439 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
440 delete $self->{_tw};
441 $self->_timeout;
442 });
443 } else {
444 delete $self->{_tw};
445 }
265} 446}
266 447
267############################################################################# 448#############################################################################
268 449
269=back 450=back
316 my $len = syswrite $self->{fh}, $self->{wbuf}; 497 my $len = syswrite $self->{fh}, $self->{wbuf};
317 498
318 if ($len >= 0) { 499 if ($len >= 0) {
319 substr $self->{wbuf}, 0, $len, ""; 500 substr $self->{wbuf}, 0, $len, "";
320 501
502 $self->{_activity} = AnyEvent->now;
503
321 $self->{on_drain}($self) 504 $self->{on_drain}($self)
322 if $self->{low_water_mark} >= length $self->{wbuf} 505 if $self->{low_water_mark} >= length $self->{wbuf}
323 && $self->{on_drain}; 506 && $self->{on_drain};
324 507
325 delete $self->{_ww} unless length $self->{wbuf}; 508 delete $self->{_ww} unless length $self->{wbuf};
326 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAWOULDBLOCK) { 509 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
327 $self->error; 510 $self->_error ($!, 1);
328 } 511 }
329 }; 512 };
330 513
331 # try to write data immediately 514 # try to write data immediately
332 $cb->(); 515 $cb->() unless $self->{autocork};
333 516
334 # if still data left in wbuf, we need to poll 517 # if still data left in wbuf, we need to poll
335 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 518 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
336 if length $self->{wbuf}; 519 if length $self->{wbuf};
337 }; 520 };
352 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 535 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
353 ->($self, @_); 536 ->($self, @_);
354 } 537 }
355 538
356 if ($self->{filter_w}) { 539 if ($self->{filter_w}) {
357 $self->{filter_w}->($self, \$_[0]); 540 $self->{filter_w}($self, \$_[0]);
358 } else { 541 } else {
359 $self->{wbuf} .= $_[0]; 542 $self->{wbuf} .= $_[0];
360 $self->_drain_wbuf; 543 $self->_drain_wbuf;
361 } 544 }
362} 545}
363 546
364=item $handle->push_write (type => @args) 547=item $handle->push_write (type => @args)
365 548
366=item $handle->unshift_write (type => @args)
367
368Instead of formatting your data yourself, you can also let this module do 549Instead of formatting your data yourself, you can also let this module do
369the job by specifying a type and type-specific arguments. 550the job by specifying a type and type-specific arguments.
370 551
371Predefined types are (if you have ideas for additional types, feel free to 552Predefined types are (if you have ideas for additional types, feel free to
372drop by and tell us): 553drop by and tell us):
376=item netstring => $string 557=item netstring => $string
377 558
378Formats the given value as netstring 559Formats the given value as netstring
379(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them). 560(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
380 561
381=back
382
383=cut 562=cut
384 563
385register_write_type netstring => sub { 564register_write_type netstring => sub {
386 my ($self, $string) = @_; 565 my ($self, $string) = @_;
387 566
388 sprintf "%d:%s,", (length $string), $string 567 sprintf "%d:%s,", (length $string), $string
568};
569
570=item packstring => $format, $data
571
572An octet string prefixed with an encoded length. The encoding C<$format>
573uses the same format as a Perl C<pack> format, but must specify a single
574integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
575optional C<!>, C<< < >> or C<< > >> modifier).
576
577=cut
578
579register_write_type packstring => sub {
580 my ($self, $format, $string) = @_;
581
582 pack "$format/a*", $string
389}; 583};
390 584
391=item json => $array_or_hashref 585=item json => $array_or_hashref
392 586
393Encodes the given hash or array reference into a JSON object. Unless you 587Encodes the given hash or array reference into a JSON object. Unless you
396 590
397JSON objects (and arrays) are self-delimiting, so you can write JSON at 591JSON objects (and arrays) are self-delimiting, so you can write JSON at
398one end of a handle and read them at the other end without using any 592one end of a handle and read them at the other end without using any
399additional framing. 593additional framing.
400 594
595The generated JSON text is guaranteed not to contain any newlines: While
596this module doesn't need delimiters after or between JSON texts to be
597able to read them, many other languages depend on that.
598
599A simple RPC protocol that interoperates easily with others is to send
600JSON arrays (or objects, although arrays are usually the better choice as
601they mimic how function argument passing works) and a newline after each
602JSON text:
603
604 $handle->push_write (json => ["method", "arg1", "arg2"]); # whatever
605 $handle->push_write ("\012");
606
607An AnyEvent::Handle receiver would simply use the C<json> read type and
608rely on the fact that the newline will be skipped as leading whitespace:
609
610 $handle->push_read (json => sub { my $array = $_[1]; ... });
611
612Other languages could read single lines terminated by a newline and pass
613this line into their JSON decoder of choice.
614
401=cut 615=cut
402 616
403register_write_type json => sub { 617register_write_type json => sub {
404 my ($self, $ref) = @_; 618 my ($self, $ref) = @_;
405 619
407 621
408 $self->{json} ? $self->{json}->encode ($ref) 622 $self->{json} ? $self->{json}->encode ($ref)
409 : JSON::encode_json ($ref) 623 : JSON::encode_json ($ref)
410}; 624};
411 625
626=item storable => $reference
627
628Freezes the given reference using L<Storable> and writes it to the
629handle. Uses the C<nfreeze> format.
630
631=cut
632
633register_write_type storable => sub {
634 my ($self, $ref) = @_;
635
636 require Storable;
637
638 pack "w/a*", Storable::nfreeze ($ref)
639};
640
641=back
642
412=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 643=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
413 644
414This function (not method) lets you add your own types to C<push_write>. 645This function (not method) lets you add your own types to C<push_write>.
415Whenever the given C<type> is used, C<push_write> will invoke the code 646Whenever the given C<type> is used, C<push_write> will invoke the code
416reference with the handle object and the remaining arguments. 647reference with the handle object and the remaining arguments.
436ways, the "simple" way, using only C<on_read> and the "complex" way, using 667ways, the "simple" way, using only C<on_read> and the "complex" way, using
437a queue. 668a queue.
438 669
439In the simple case, you just install an C<on_read> callback and whenever 670In the simple case, you just install an C<on_read> callback and whenever
440new data arrives, it will be called. You can then remove some data (if 671new data arrives, it will be called. You can then remove some data (if
441enough is there) from the read buffer (C<< $handle->rbuf >>) if you want 672enough is there) from the read buffer (C<< $handle->rbuf >>). Or you cna
442or not. 673leave the data there if you want to accumulate more (e.g. when only a
674partial message has been received so far).
443 675
444In the more complex case, you want to queue multiple callbacks. In this 676In the more complex case, you want to queue multiple callbacks. In this
445case, AnyEvent::Handle will call the first queued callback each time new 677case, AnyEvent::Handle will call the first queued callback each time new
446data arrives and removes it when it has done its job (see C<push_read>, 678data arrives (also the first time it is queued) and removes it when it has
447below). 679done its job (see C<push_read>, below).
448 680
449This way you can, for example, push three line-reads, followed by reading 681This way you can, for example, push three line-reads, followed by reading
450a chunk of data, and AnyEvent::Handle will execute them in order. 682a chunk of data, and AnyEvent::Handle will execute them in order.
451 683
452Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by 684Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
453the specified number of bytes which give an XML datagram. 685the specified number of bytes which give an XML datagram.
454 686
455 # in the default state, expect some header bytes 687 # in the default state, expect some header bytes
456 $handle->on_read (sub { 688 $handle->on_read (sub {
457 # some data is here, now queue the length-header-read (4 octets) 689 # some data is here, now queue the length-header-read (4 octets)
458 shift->unshift_read_chunk (4, sub { 690 shift->unshift_read (chunk => 4, sub {
459 # header arrived, decode 691 # header arrived, decode
460 my $len = unpack "N", $_[1]; 692 my $len = unpack "N", $_[1];
461 693
462 # now read the payload 694 # now read the payload
463 shift->unshift_read_chunk ($len, sub { 695 shift->unshift_read (chunk => $len, sub {
464 my $xml = $_[1]; 696 my $xml = $_[1];
465 # handle xml 697 # handle xml
466 }); 698 });
467 }); 699 });
468 }); 700 });
469 701
470Example 2: Implement a client for a protocol that replies either with 702Example 2: Implement a client for a protocol that replies either with "OK"
471"OK" and another line or "ERROR" for one request, and 64 bytes for the 703and another line or "ERROR" for the first request that is sent, and 64
472second request. Due tot he availability of a full queue, we can just 704bytes for the second request. Due to the availability of a queue, we can
473pipeline sending both requests and manipulate the queue as necessary in 705just pipeline sending both requests and manipulate the queue as necessary
474the callbacks: 706in the callbacks.
475 707
476 # request one 708When the first callback is called and sees an "OK" response, it will
709C<unshift> another line-read. This line-read will be queued I<before> the
71064-byte chunk callback.
711
712 # request one, returns either "OK + extra line" or "ERROR"
477 $handle->push_write ("request 1\015\012"); 713 $handle->push_write ("request 1\015\012");
478 714
479 # we expect "ERROR" or "OK" as response, so push a line read 715 # we expect "ERROR" or "OK" as response, so push a line read
480 $handle->push_read_line (sub { 716 $handle->push_read (line => sub {
481 # if we got an "OK", we have to _prepend_ another line, 717 # if we got an "OK", we have to _prepend_ another line,
482 # so it will be read before the second request reads its 64 bytes 718 # so it will be read before the second request reads its 64 bytes
483 # which are already in the queue when this callback is called 719 # which are already in the queue when this callback is called
484 # we don't do this in case we got an error 720 # we don't do this in case we got an error
485 if ($_[1] eq "OK") { 721 if ($_[1] eq "OK") {
486 $_[0]->unshift_read_line (sub { 722 $_[0]->unshift_read (line => sub {
487 my $response = $_[1]; 723 my $response = $_[1];
488 ... 724 ...
489 }); 725 });
490 } 726 }
491 }); 727 });
492 728
493 # request two 729 # request two, simply returns 64 octets
494 $handle->push_write ("request 2\015\012"); 730 $handle->push_write ("request 2\015\012");
495 731
496 # simply read 64 bytes, always 732 # simply read 64 bytes, always
497 $handle->push_read_chunk (64, sub { 733 $handle->push_read (chunk => 64, sub {
498 my $response = $_[1]; 734 my $response = $_[1];
499 ... 735 ...
500 }); 736 });
501 737
502=over 4 738=over 4
503 739
504=cut 740=cut
505 741
506sub _drain_rbuf { 742sub _drain_rbuf {
507 my ($self) = @_; 743 my ($self) = @_;
744
745 local $self->{_in_drain} = 1;
508 746
509 if ( 747 if (
510 defined $self->{rbuf_max} 748 defined $self->{rbuf_max}
511 && $self->{rbuf_max} < length $self->{rbuf} 749 && $self->{rbuf_max} < length $self->{rbuf}
512 ) { 750 ) {
513 $! = &Errno::ENOSPC; 751 $self->_error (&Errno::ENOSPC, 1), return;
514 $self->error;
515 } 752 }
516 753
517 return if $self->{in_drain}; 754 while () {
518 local $self->{in_drain} = 1;
519
520 while (my $len = length $self->{rbuf}) { 755 my $len = length $self->{rbuf};
521 no strict 'refs'; 756
522 if (my $cb = shift @{ $self->{_queue} }) { 757 if (my $cb = shift @{ $self->{_queue} }) {
523 unless ($cb->($self)) { 758 unless ($cb->($self)) {
524 if ($self->{_eof}) { 759 if ($self->{_eof}) {
525 # no progress can be made (not enough data and no data forthcoming) 760 # no progress can be made (not enough data and no data forthcoming)
526 $! = &Errno::EPIPE; 761 $self->_error (&Errno::EPIPE, 1), return;
527 $self->error;
528 } 762 }
529 763
530 unshift @{ $self->{_queue} }, $cb; 764 unshift @{ $self->{_queue} }, $cb;
531 return; 765 last;
532 } 766 }
533 } elsif ($self->{on_read}) { 767 } elsif ($self->{on_read}) {
768 last unless $len;
769
534 $self->{on_read}($self); 770 $self->{on_read}($self);
535 771
536 if ( 772 if (
537 $self->{_eof} # if no further data will arrive
538 && $len == length $self->{rbuf} # and no data has been consumed 773 $len == length $self->{rbuf} # if no data has been consumed
539 && !@{ $self->{_queue} } # and the queue is still empty 774 && !@{ $self->{_queue} } # and the queue is still empty
540 && $self->{on_read} # and we still want to read data 775 && $self->{on_read} # but we still have on_read
541 ) { 776 ) {
777 # no further data will arrive
542 # then no progress can be made 778 # so no progress can be made
543 $! = &Errno::EPIPE; 779 $self->_error (&Errno::EPIPE, 1), return
544 $self->error; 780 if $self->{_eof};
781
782 last; # more data might arrive
545 } 783 }
546 } else { 784 } else {
547 # read side becomes idle 785 # read side becomes idle
548 delete $self->{_rw}; 786 delete $self->{_rw};
549 return; 787 last;
550 } 788 }
551 } 789 }
552 790
553 if ($self->{_eof}) { 791 if ($self->{_eof}) {
554 $self->_shutdown; 792 if ($self->{on_eof}) {
555 $self->{on_eof}($self) 793 $self->{on_eof}($self)
556 if $self->{on_eof}; 794 } else {
795 $self->_error (0, 1);
796 }
797 }
798
799 # may need to restart read watcher
800 unless ($self->{_rw}) {
801 $self->start_read
802 if $self->{on_read} || @{ $self->{_queue} };
557 } 803 }
558} 804}
559 805
560=item $handle->on_read ($cb) 806=item $handle->on_read ($cb)
561 807
567 813
568sub on_read { 814sub on_read {
569 my ($self, $cb) = @_; 815 my ($self, $cb) = @_;
570 816
571 $self->{on_read} = $cb; 817 $self->{on_read} = $cb;
818 $self->_drain_rbuf if $cb && !$self->{_in_drain};
572} 819}
573 820
574=item $handle->rbuf 821=item $handle->rbuf
575 822
576Returns the read buffer (as a modifiable lvalue). 823Returns the read buffer (as a modifiable lvalue).
625 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 872 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
626 ->($self, $cb, @_); 873 ->($self, $cb, @_);
627 } 874 }
628 875
629 push @{ $self->{_queue} }, $cb; 876 push @{ $self->{_queue} }, $cb;
630 $self->_drain_rbuf; 877 $self->_drain_rbuf unless $self->{_in_drain};
631} 878}
632 879
633sub unshift_read { 880sub unshift_read {
634 my $self = shift; 881 my $self = shift;
635 my $cb = pop; 882 my $cb = pop;
641 ->($self, $cb, @_); 888 ->($self, $cb, @_);
642 } 889 }
643 890
644 891
645 unshift @{ $self->{_queue} }, $cb; 892 unshift @{ $self->{_queue} }, $cb;
646 $self->_drain_rbuf; 893 $self->_drain_rbuf unless $self->{_in_drain};
647} 894}
648 895
649=item $handle->push_read (type => @args, $cb) 896=item $handle->push_read (type => @args, $cb)
650 897
651=item $handle->unshift_read (type => @args, $cb) 898=item $handle->unshift_read (type => @args, $cb)
681 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 928 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
682 1 929 1
683 } 930 }
684}; 931};
685 932
686# compatibility with older API
687sub push_read_chunk {
688 $_[0]->push_read (chunk => $_[1], $_[2]);
689}
690
691sub unshift_read_chunk {
692 $_[0]->unshift_read (chunk => $_[1], $_[2]);
693}
694
695=item line => [$eol, ]$cb->($handle, $line, $eol) 933=item line => [$eol, ]$cb->($handle, $line, $eol)
696 934
697The callback will be called only once a full line (including the end of 935The callback will be called only once a full line (including the end of
698line marker, C<$eol>) has been read. This line (excluding the end of line 936line marker, C<$eol>) has been read. This line (excluding the end of line
699marker) will be passed to the callback as second argument (C<$line>), and 937marker) will be passed to the callback as second argument (C<$line>), and
714=cut 952=cut
715 953
716register_read_type line => sub { 954register_read_type line => sub {
717 my ($self, $cb, $eol) = @_; 955 my ($self, $cb, $eol) = @_;
718 956
719 $eol = qr|(\015?\012)| if @_ < 3; 957 if (@_ < 3) {
720 $eol = quotemeta $eol unless ref $eol; 958 # this is more than twice as fast as the generic code below
721 $eol = qr|^(.*?)($eol)|s;
722
723 sub { 959 sub {
724 $_[0]{rbuf} =~ s/$eol// or return; 960 $_[0]{rbuf} =~ s/^([^\015\012]*)(\015?\012)// or return;
725 961
726 $cb->($_[0], $1, $2); 962 $cb->($_[0], $1, $2);
727 1
728 }
729};
730
731# compatibility with older API
732sub push_read_line {
733 my $self = shift;
734 $self->push_read (line => @_);
735}
736
737sub unshift_read_line {
738 my $self = shift;
739 $self->unshift_read (line => @_);
740}
741
742=item netstring => $cb->($handle, $string)
743
744A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
745
746Throws an error with C<$!> set to EBADMSG on format violations.
747
748=cut
749
750register_read_type netstring => sub {
751 my ($self, $cb) = @_;
752
753 sub {
754 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
755 if ($_[0]{rbuf} =~ /[^0-9]/) {
756 $! = &Errno::EBADMSG;
757 $self->error;
758 } 963 1
759 return;
760 } 964 }
965 } else {
966 $eol = quotemeta $eol unless ref $eol;
967 $eol = qr|^(.*?)($eol)|s;
761 968
762 my $len = $1; 969 sub {
970 $_[0]{rbuf} =~ s/$eol// or return;
763 971
764 $self->unshift_read (chunk => $len, sub { 972 $cb->($_[0], $1, $2);
765 my $string = $_[1];
766 $_[0]->unshift_read (chunk => 1, sub {
767 if ($_[1] eq ",") {
768 $cb->($_[0], $string);
769 } else {
770 $! = &Errno::EBADMSG;
771 $self->error;
772 }
773 }); 973 1
774 }); 974 }
775
776 1
777 } 975 }
778}; 976};
779 977
780=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 978=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
781 979
833 return 1; 1031 return 1;
834 } 1032 }
835 1033
836 # reject 1034 # reject
837 if ($reject && $$rbuf =~ $reject) { 1035 if ($reject && $$rbuf =~ $reject) {
838 $! = &Errno::EBADMSG; 1036 $self->_error (&Errno::EBADMSG);
839 $self->error;
840 } 1037 }
841 1038
842 # skip 1039 # skip
843 if ($skip && $$rbuf =~ $skip) { 1040 if ($skip && $$rbuf =~ $skip) {
844 $data .= substr $$rbuf, 0, $+[0], ""; 1041 $data .= substr $$rbuf, 0, $+[0], "";
846 1043
847 () 1044 ()
848 } 1045 }
849}; 1046};
850 1047
1048=item netstring => $cb->($handle, $string)
1049
1050A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
1051
1052Throws an error with C<$!> set to EBADMSG on format violations.
1053
1054=cut
1055
1056register_read_type netstring => sub {
1057 my ($self, $cb) = @_;
1058
1059 sub {
1060 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
1061 if ($_[0]{rbuf} =~ /[^0-9]/) {
1062 $self->_error (&Errno::EBADMSG);
1063 }
1064 return;
1065 }
1066
1067 my $len = $1;
1068
1069 $self->unshift_read (chunk => $len, sub {
1070 my $string = $_[1];
1071 $_[0]->unshift_read (chunk => 1, sub {
1072 if ($_[1] eq ",") {
1073 $cb->($_[0], $string);
1074 } else {
1075 $self->_error (&Errno::EBADMSG);
1076 }
1077 });
1078 });
1079
1080 1
1081 }
1082};
1083
1084=item packstring => $format, $cb->($handle, $string)
1085
1086An octet string prefixed with an encoded length. The encoding C<$format>
1087uses the same format as a Perl C<pack> format, but must specify a single
1088integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
1089optional C<!>, C<< < >> or C<< > >> modifier).
1090
1091DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
1092
1093Example: read a block of data prefixed by its length in BER-encoded
1094format (very efficient).
1095
1096 $handle->push_read (packstring => "w", sub {
1097 my ($handle, $data) = @_;
1098 });
1099
1100=cut
1101
1102register_read_type packstring => sub {
1103 my ($self, $cb, $format) = @_;
1104
1105 sub {
1106 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1107 defined (my $len = eval { unpack $format, $_[0]{rbuf} })
1108 or return;
1109
1110 $format = length pack $format, $len;
1111
1112 # bypass unshift if we already have the remaining chunk
1113 if ($format + $len <= length $_[0]{rbuf}) {
1114 my $data = substr $_[0]{rbuf}, $format, $len;
1115 substr $_[0]{rbuf}, 0, $format + $len, "";
1116 $cb->($_[0], $data);
1117 } else {
1118 # remove prefix
1119 substr $_[0]{rbuf}, 0, $format, "";
1120
1121 # read remaining chunk
1122 $_[0]->unshift_read (chunk => $len, $cb);
1123 }
1124
1125 1
1126 }
1127};
1128
851=item json => $cb->($handle, $hash_or_arrayref) 1129=item json => $cb->($handle, $hash_or_arrayref)
852 1130
853Reads a JSON object or array, decodes it and passes it to the callback. 1131Reads a JSON object or array, decodes it and passes it to the callback.
854 1132
855If a C<json> object was passed to the constructor, then that will be used 1133If a C<json> object was passed to the constructor, then that will be used
8592.09 (and JSON::XS version 2.2) and above. You have to provide a 11372.09 (and JSON::XS version 2.2) and above. You have to provide a
860dependency on your own: this module will load the JSON module, but 1138dependency on your own: this module will load the JSON module, but
861AnyEvent does not depend on it itself. 1139AnyEvent does not depend on it itself.
862 1140
863Since JSON texts are fully self-delimiting, the C<json> read and write 1141Since JSON texts are fully self-delimiting, the C<json> read and write
864types are an ideal simple RPC protocol: just exchange JSON datagrams. 1142types are an ideal simple RPC protocol: just exchange JSON datagrams. See
1143the C<json> write type description, above, for an actual example.
865 1144
866=cut 1145=cut
867 1146
868register_read_type json => sub { 1147register_read_type json => sub {
869 my ($self, $cb, $accept, $reject, $skip) = @_; 1148 my ($self, $cb) = @_;
870 1149
871 require JSON; 1150 require JSON;
872 1151
873 my $data; 1152 my $data;
874 my $rbuf = \$self->{rbuf}; 1153 my $rbuf = \$self->{rbuf};
875 1154
876 my $json = $self->{json} ||= JSON::XS->new->utf8; 1155 my $json = $self->{json} ||= JSON->new->utf8;
877 1156
878 sub { 1157 sub {
879 my $ref = $json->incr_parse ($self->{rbuf}); 1158 my $ref = $json->incr_parse ($self->{rbuf});
880 1159
881 if ($ref) { 1160 if ($ref) {
889 () 1168 ()
890 } 1169 }
891 } 1170 }
892}; 1171};
893 1172
1173=item storable => $cb->($handle, $ref)
1174
1175Deserialises a L<Storable> frozen representation as written by the
1176C<storable> write type (BER-encoded length prefix followed by nfreeze'd
1177data).
1178
1179Raises C<EBADMSG> error if the data could not be decoded.
1180
1181=cut
1182
1183register_read_type storable => sub {
1184 my ($self, $cb) = @_;
1185
1186 require Storable;
1187
1188 sub {
1189 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1190 defined (my $len = eval { unpack "w", $_[0]{rbuf} })
1191 or return;
1192
1193 my $format = length pack "w", $len;
1194
1195 # bypass unshift if we already have the remaining chunk
1196 if ($format + $len <= length $_[0]{rbuf}) {
1197 my $data = substr $_[0]{rbuf}, $format, $len;
1198 substr $_[0]{rbuf}, 0, $format + $len, "";
1199 $cb->($_[0], Storable::thaw ($data));
1200 } else {
1201 # remove prefix
1202 substr $_[0]{rbuf}, 0, $format, "";
1203
1204 # read remaining chunk
1205 $_[0]->unshift_read (chunk => $len, sub {
1206 if (my $ref = eval { Storable::thaw ($_[1]) }) {
1207 $cb->($_[0], $ref);
1208 } else {
1209 $self->_error (&Errno::EBADMSG);
1210 }
1211 });
1212 }
1213
1214 1
1215 }
1216};
1217
894=back 1218=back
895 1219
896=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args) 1220=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
897 1221
898This function (not method) lets you add your own types to C<push_read>. 1222This function (not method) lets you add your own types to C<push_read>.
916=item $handle->stop_read 1240=item $handle->stop_read
917 1241
918=item $handle->start_read 1242=item $handle->start_read
919 1243
920In rare cases you actually do not want to read anything from the 1244In rare cases you actually do not want to read anything from the
921socket. In this case you can call C<stop_read>. Neither C<on_read> no 1245socket. In this case you can call C<stop_read>. Neither C<on_read> nor
922any queued callbacks will be executed then. To start reading again, call 1246any queued callbacks will be executed then. To start reading again, call
923C<start_read>. 1247C<start_read>.
1248
1249Note that AnyEvent::Handle will automatically C<start_read> for you when
1250you change the C<on_read> callback or push/unshift a read callback, and it
1251will automatically C<stop_read> for you when neither C<on_read> is set nor
1252there are any read requests in the queue.
924 1253
925=cut 1254=cut
926 1255
927sub stop_read { 1256sub stop_read {
928 my ($self) = @_; 1257 my ($self) = @_;
939 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1268 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
940 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1269 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
941 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1270 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
942 1271
943 if ($len > 0) { 1272 if ($len > 0) {
1273 $self->{_activity} = AnyEvent->now;
1274
944 $self->{filter_r} 1275 $self->{filter_r}
945 ? $self->{filter_r}->($self, $rbuf) 1276 ? $self->{filter_r}($self, $rbuf)
946 : $self->_drain_rbuf; 1277 : $self->{_in_drain} || $self->_drain_rbuf;
947 1278
948 } elsif (defined $len) { 1279 } elsif (defined $len) {
949 delete $self->{_rw}; 1280 delete $self->{_rw};
950 $self->{_eof} = 1; 1281 $self->{_eof} = 1;
951 $self->_drain_rbuf; 1282 $self->_drain_rbuf unless $self->{_in_drain};
952 1283
953 } elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) { 1284 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
954 return $self->error; 1285 return $self->_error ($!, 1);
955 } 1286 }
956 }); 1287 });
957 } 1288 }
958} 1289}
959 1290
960sub _dotls { 1291sub _dotls {
961 my ($self) = @_; 1292 my ($self) = @_;
1293
1294 my $buf;
962 1295
963 if (length $self->{_tls_wbuf}) { 1296 if (length $self->{_tls_wbuf}) {
964 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1297 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
965 substr $self->{_tls_wbuf}, 0, $len, ""; 1298 substr $self->{_tls_wbuf}, 0, $len, "";
966 } 1299 }
967 } 1300 }
968 1301
969 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1302 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
970 $self->{wbuf} .= $buf; 1303 $self->{wbuf} .= $buf;
971 $self->_drain_wbuf; 1304 $self->_drain_wbuf;
972 } 1305 }
973 1306
974 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1307 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1308 if (length $buf) {
975 $self->{rbuf} .= $buf; 1309 $self->{rbuf} .= $buf;
976 $self->_drain_rbuf; 1310 $self->_drain_rbuf unless $self->{_in_drain};
1311 } else {
1312 # let's treat SSL-eof as we treat normal EOF
1313 $self->{_eof} = 1;
1314 $self->_shutdown;
1315 return;
1316 }
977 } 1317 }
978 1318
979 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1319 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
980 1320
981 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1321 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
982 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1322 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
983 $self->error; 1323 return $self->_error ($!, 1);
984 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1324 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
985 $! = &Errno::EIO; 1325 return $self->_error (&Errno::EIO, 1);
986 $self->error;
987 } 1326 }
988 1327
989 # all others are fine for our purposes 1328 # all others are fine for our purposes
990 } 1329 }
991} 1330}
1006call and can be used or changed to your liking. Note that the handshake 1345call and can be used or changed to your liking. Note that the handshake
1007might have already started when this function returns. 1346might have already started when this function returns.
1008 1347
1009=cut 1348=cut
1010 1349
1011# TODO: maybe document...
1012sub starttls { 1350sub starttls {
1013 my ($self, $ssl, $ctx) = @_; 1351 my ($self, $ssl, $ctx) = @_;
1014 1352
1015 $self->stoptls; 1353 $self->stoptls;
1016 1354
1069 1407
1070sub DESTROY { 1408sub DESTROY {
1071 my $self = shift; 1409 my $self = shift;
1072 1410
1073 $self->stoptls; 1411 $self->stoptls;
1412
1413 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1414
1415 if ($linger && length $self->{wbuf}) {
1416 my $fh = delete $self->{fh};
1417 my $wbuf = delete $self->{wbuf};
1418
1419 my @linger;
1420
1421 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {
1422 my $len = syswrite $fh, $wbuf, length $wbuf;
1423
1424 if ($len > 0) {
1425 substr $wbuf, 0, $len, "";
1426 } else {
1427 @linger = (); # end
1428 }
1429 });
1430 push @linger, AnyEvent->timer (after => $linger, cb => sub {
1431 @linger = ();
1432 });
1433 }
1074} 1434}
1075 1435
1076=item AnyEvent::Handle::TLS_CTX 1436=item AnyEvent::Handle::TLS_CTX
1077 1437
1078This function creates and returns the Net::SSLeay::CTX object used by 1438This function creates and returns the Net::SSLeay::CTX object used by
1120=over 4 1480=over 4
1121 1481
1122=item * all constructor arguments become object members. 1482=item * all constructor arguments become object members.
1123 1483
1124At least initially, when you pass a C<tls>-argument to the constructor it 1484At least initially, when you pass a C<tls>-argument to the constructor it
1125will end up in C<< $handle->{tls} >>. Those members might be changes or 1485will end up in C<< $handle->{tls} >>. Those members might be changed or
1126mutated later on (for example C<tls> will hold the TLS connection object). 1486mutated later on (for example C<tls> will hold the TLS connection object).
1127 1487
1128=item * other object member names are prefixed with an C<_>. 1488=item * other object member names are prefixed with an C<_>.
1129 1489
1130All object members not explicitly documented (internal use) are prefixed 1490All object members not explicitly documented (internal use) are prefixed

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines