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.34 by root, Mon May 26 03:27:52 2008 UTC vs.
Revision 1.69 by root, Sun Jun 15 21:44:56 2008 UTC

2 2
3no warnings; 3no warnings;
4use strict; 4use strict;
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.151;
20 20
21=head1 SYNOPSIS 21=head1 SYNOPSIS
22 22
23 use AnyEvent; 23 use AnyEvent;
24 use AnyEvent::Handle; 24 use AnyEvent::Handle;
73The filehandle this L<AnyEvent::Handle> object will operate on. 73The filehandle this L<AnyEvent::Handle> object will operate on.
74 74
75NOTE: The filehandle will be set to non-blocking (using 75NOTE: The filehandle will be set to non-blocking (using
76AnyEvent::Util::fh_nonblocking). 76AnyEvent::Util::fh_nonblocking).
77 77
78=item on_eof => $cb->($self) 78=item on_eof => $cb->($handle)
79 79
80Set the callback to be called on EOF. 80Set the callback to be called when an end-of-file condition is detcted,
81i.e. in the case of a socket, when the other side has closed the
82connection cleanly.
81 83
82While not mandatory, it is highly recommended to set an eof callback, 84While not mandatory, it is highly recommended to set an eof callback,
83otherwise you might end up with a closed socket while you are still 85otherwise you might end up with a closed socket while you are still
84waiting for data. 86waiting for data.
85 87
86=item on_error => $cb->($self) 88=item on_error => $cb->($handle, $fatal)
87 89
88This is the fatal error callback, that is called when, well, a fatal error 90This is the error callback, which is called when, well, some error
89occurs, such as not being able to resolve the hostname, failure to connect 91occured, such as not being able to resolve the hostname, failure to
90or a read error. 92connect or a read error.
91 93
92The object will not be in a usable state when this callback has been 94Some errors are fatal (which is indicated by C<$fatal> being true). On
93called. 95fatal errors the handle object will be shut down and will not be
96usable. Non-fatal errors can be retried by simply returning, but it is
97recommended to simply ignore this parameter and instead abondon the handle
98object when this callback is invoked.
94 99
95On callback entrance, the value of C<$!> contains the operating system 100On callback entrance, the value of C<$!> contains the operating system
96error (or C<ENOSPC>, C<EPIPE> or C<EBADMSG>). 101error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>).
97 102
98While not mandatory, it is I<highly> recommended to set this callback, as 103While not mandatory, it is I<highly> recommended to set this callback, as
99you will not be notified of errors otherwise. The default simply calls 104you will not be notified of errors otherwise. The default simply calls
100die. 105C<croak>.
101 106
102=item on_read => $cb->($self) 107=item on_read => $cb->($handle)
103 108
104This sets the default read callback, which is called when data arrives 109This sets the default read callback, which is called when data arrives
105and no read request is in the queue. 110and no read request is in the queue (unlike read queue callbacks, this
111callback will only be called when at least one octet of data is in the
112read buffer).
106 113
107To access (and remove data from) the read buffer, use the C<< ->rbuf >> 114To access (and remove data from) the read buffer, use the C<< ->rbuf >>
108method or access the C<$self->{rbuf}> member directly. 115method or access the C<$handle->{rbuf}> member directly.
109 116
110When an EOF condition is detected then AnyEvent::Handle will first try to 117When an EOF condition is detected then AnyEvent::Handle will first try to
111feed all the remaining data to the queued callbacks and C<on_read> before 118feed all the remaining data to the queued callbacks and C<on_read> before
112calling the C<on_eof> callback. If no progress can be made, then a fatal 119calling the C<on_eof> callback. If no progress can be made, then a fatal
113error will be raised (with C<$!> set to C<EPIPE>). 120error will be raised (with C<$!> set to C<EPIPE>).
114 121
115=item on_drain => $cb->() 122=item on_drain => $cb->($handle)
116 123
117This sets the callback that is called when the write buffer becomes empty 124This sets the callback that is called when the write buffer becomes empty
118(or when the callback is set and the buffer is empty already). 125(or when the callback is set and the buffer is empty already).
119 126
120To append to the write buffer, use the C<< ->push_write >> method. 127To append to the write buffer, use the C<< ->push_write >> method.
128
129This callback is useful when you don't want to put all of your write data
130into the queue at once, for example, when you want to write the contents
131of some file to the socket you might not want to read the whole file into
132memory and push it into the queue, but instead only read more data from
133the file when the write queue becomes empty.
134
135=item timeout => $fractional_seconds
136
137If non-zero, then this enables an "inactivity" timeout: whenever this many
138seconds pass without a successful read or write on the underlying file
139handle, the C<on_timeout> callback will be invoked (and if that one is
140missing, an C<ETIMEDOUT> error will be raised).
141
142Note that timeout processing is also active when you currently do not have
143any outstanding read or write requests: If you plan to keep the connection
144idle then you should disable the timout temporarily or ignore the timeout
145in the C<on_timeout> callback.
146
147Zero (the default) disables this timeout.
148
149=item on_timeout => $cb->($handle)
150
151Called whenever the inactivity timeout passes. If you return from this
152callback, then the timeout will be reset as if some activity had happened,
153so this condition is not fatal in any way.
121 154
122=item rbuf_max => <bytes> 155=item rbuf_max => <bytes>
123 156
124If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>) 157If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>)
125when the read buffer ever (strictly) exceeds this size. This is useful to 158when the read buffer ever (strictly) exceeds this size. This is useful to
132isn't finished). 165isn't finished).
133 166
134=item read_size => <bytes> 167=item read_size => <bytes>
135 168
136The default read block size (the amount of bytes this module will try to read 169The default read block size (the amount of bytes this module will try to read
137on each [loop iteration). Default: C<4096>. 170during each (loop iteration). Default: C<8192>.
138 171
139=item low_water_mark => <bytes> 172=item low_water_mark => <bytes>
140 173
141Sets the amount of bytes (default: C<0>) that make up an "empty" write 174Sets the amount of bytes (default: C<0>) that make up an "empty" write
142buffer: If the write reaches this size or gets even samller it is 175buffer: If the write reaches this size or gets even samller it is
143considered empty. 176considered empty.
177
178=item linger => <seconds>
179
180If non-zero (default: C<3600>), then the destructor of the
181AnyEvent::Handle object will check wether there is still outstanding write
182data and will install a watcher that will write out this data. No errors
183will be reported (this mostly matches how the operating system treats
184outstanding data at socket close time).
185
186This will not work for partial TLS data that could not yet been
187encoded. This data will be lost.
144 188
145=item tls => "accept" | "connect" | Net::SSLeay::SSL object 189=item tls => "accept" | "connect" | Net::SSLeay::SSL object
146 190
147When this parameter is given, it enables TLS (SSL) mode, that means it 191When this parameter is given, it enables TLS (SSL) mode, that means it
148will start making tls handshake and will transparently encrypt/decrypt 192will start making tls handshake and will transparently encrypt/decrypt
165 209
166Use the given Net::SSLeay::CTX object to create the new TLS connection 210Use the given Net::SSLeay::CTX object to create the new TLS connection
167(unless a connection object was specified directly). If this parameter is 211(unless a connection object was specified directly). If this parameter is
168missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>. 212missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>.
169 213
214=item json => JSON or JSON::XS object
215
216This is the json coder object used by the C<json> read and write types.
217
218If you don't supply it, then AnyEvent::Handle will create and use a
219suitable one, which will write and expect UTF-8 encoded JSON texts.
220
221Note that you are responsible to depend on the JSON module if you want to
222use this functionality, as AnyEvent does not have a dependency itself.
223
224=item filter_r => $cb
225
226=item filter_w => $cb
227
228These exist, but are undocumented at this time.
229
170=back 230=back
171 231
172=cut 232=cut
173 233
174sub new { 234sub new {
183 if ($self->{tls}) { 243 if ($self->{tls}) {
184 require Net::SSLeay; 244 require Net::SSLeay;
185 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 245 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
186 } 246 }
187 247
188 $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; 248 $self->{_activity} = AnyEvent->now;
189 $self->on_error (delete $self->{on_error}) if $self->{on_error}; 249 $self->_timeout;
250
190 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 251 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
191 $self->on_read (delete $self->{on_read} ) if $self->{on_read};
192 252
193 $self->start_read; 253 $self->start_read
254 if $self->{on_read};
194 255
195 $self 256 $self
196} 257}
197 258
198sub _shutdown { 259sub _shutdown {
199 my ($self) = @_; 260 my ($self) = @_;
200 261
262 delete $self->{_tw};
201 delete $self->{rw}; 263 delete $self->{_rw};
202 delete $self->{ww}; 264 delete $self->{_ww};
203 delete $self->{fh}; 265 delete $self->{fh};
204}
205 266
267 $self->stoptls;
268}
269
206sub error { 270sub _error {
207 my ($self) = @_; 271 my ($self, $errno, $fatal) = @_;
208 272
209 {
210 local $!;
211 $self->_shutdown; 273 $self->_shutdown
212 } 274 if $fatal;
275
276 $! = $errno;
213 277
214 if ($self->{on_error}) { 278 if ($self->{on_error}) {
215 $self->{on_error}($self); 279 $self->{on_error}($self, $fatal);
216 } else { 280 } else {
217 Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; 281 Carp::croak "AnyEvent::Handle uncaught error: $!";
218 } 282 }
219} 283}
220 284
221=item $fh = $handle->fh 285=item $fh = $handle->fh
222 286
223This method returns the file handle of the L<AnyEvent::Handle> object. 287This method returns the file handle of the L<AnyEvent::Handle> object.
224 288
225=cut 289=cut
226 290
227sub fh { $_[0]->{fh} } 291sub fh { $_[0]{fh} }
228 292
229=item $handle->on_error ($cb) 293=item $handle->on_error ($cb)
230 294
231Replace the current C<on_error> callback (see the C<on_error> constructor argument). 295Replace the current C<on_error> callback (see the C<on_error> constructor argument).
232 296
242 306
243=cut 307=cut
244 308
245sub on_eof { 309sub on_eof {
246 $_[0]{on_eof} = $_[1]; 310 $_[0]{on_eof} = $_[1];
311}
312
313=item $handle->on_timeout ($cb)
314
315Replace the current C<on_timeout> callback, or disables the callback
316(but not the timeout) if C<$cb> = C<undef>. See C<timeout> constructor
317argument.
318
319=cut
320
321sub on_timeout {
322 $_[0]{on_timeout} = $_[1];
323}
324
325#############################################################################
326
327=item $handle->timeout ($seconds)
328
329Configures (or disables) the inactivity timeout.
330
331=cut
332
333sub timeout {
334 my ($self, $timeout) = @_;
335
336 $self->{timeout} = $timeout;
337 $self->_timeout;
338}
339
340# reset the timeout watcher, as neccessary
341# also check for time-outs
342sub _timeout {
343 my ($self) = @_;
344
345 if ($self->{timeout}) {
346 my $NOW = AnyEvent->now;
347
348 # when would the timeout trigger?
349 my $after = $self->{_activity} + $self->{timeout} - $NOW;
350
351 # now or in the past already?
352 if ($after <= 0) {
353 $self->{_activity} = $NOW;
354
355 if ($self->{on_timeout}) {
356 $self->{on_timeout}($self);
357 } else {
358 $self->_error (&Errno::ETIMEDOUT);
359 }
360
361 # callback could have changed timeout value, optimise
362 return unless $self->{timeout};
363
364 # calculate new after
365 $after = $self->{timeout};
366 }
367
368 Scalar::Util::weaken $self;
369 return unless $self; # ->error could have destroyed $self
370
371 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
372 delete $self->{_tw};
373 $self->_timeout;
374 });
375 } else {
376 delete $self->{_tw};
377 }
247} 378}
248 379
249############################################################################# 380#############################################################################
250 381
251=back 382=back
288=cut 419=cut
289 420
290sub _drain_wbuf { 421sub _drain_wbuf {
291 my ($self) = @_; 422 my ($self) = @_;
292 423
293 if (!$self->{ww} && length $self->{wbuf}) { 424 if (!$self->{_ww} && length $self->{wbuf}) {
425
294 Scalar::Util::weaken $self; 426 Scalar::Util::weaken $self;
427
295 my $cb = sub { 428 my $cb = sub {
296 my $len = syswrite $self->{fh}, $self->{wbuf}; 429 my $len = syswrite $self->{fh}, $self->{wbuf};
297 430
298 if ($len >= 0) { 431 if ($len >= 0) {
299 substr $self->{wbuf}, 0, $len, ""; 432 substr $self->{wbuf}, 0, $len, "";
433
434 $self->{_activity} = AnyEvent->now;
300 435
301 $self->{on_drain}($self) 436 $self->{on_drain}($self)
302 if $self->{low_water_mark} >= length $self->{wbuf} 437 if $self->{low_water_mark} >= length $self->{wbuf}
303 && $self->{on_drain}; 438 && $self->{on_drain};
304 439
305 delete $self->{ww} unless length $self->{wbuf}; 440 delete $self->{_ww} unless length $self->{wbuf};
306 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAWOULDBLOCK) { 441 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
307 $self->error; 442 $self->_error ($!, 1);
308 } 443 }
309 }; 444 };
310 445
446 # try to write data immediately
447 $cb->();
448
449 # if still data left in wbuf, we need to poll
311 $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb); 450 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
312 451 if length $self->{wbuf};
313 $cb->($self);
314 }; 452 };
315} 453}
316 454
317our %WH; 455our %WH;
318 456
329 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 467 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
330 ->($self, @_); 468 ->($self, @_);
331 } 469 }
332 470
333 if ($self->{filter_w}) { 471 if ($self->{filter_w}) {
334 $self->{filter_w}->($self, \$_[0]); 472 $self->{filter_w}($self, \$_[0]);
335 } else { 473 } else {
336 $self->{wbuf} .= $_[0]; 474 $self->{wbuf} .= $_[0];
337 $self->_drain_wbuf; 475 $self->_drain_wbuf;
338 } 476 }
339} 477}
340 478
341=item $handle->push_write (type => @args) 479=item $handle->push_write (type => @args)
342 480
343=item $handle->unshift_write (type => @args)
344
345Instead of formatting your data yourself, you can also let this module do 481Instead of formatting your data yourself, you can also let this module do
346the job by specifying a type and type-specific arguments. 482the job by specifying a type and type-specific arguments.
347 483
348Predefined types are (if you have ideas for additional types, feel free to 484Predefined types are (if you have ideas for additional types, feel free to
349drop by and tell us): 485drop by and tell us):
353=item netstring => $string 489=item netstring => $string
354 490
355Formats the given value as netstring 491Formats the given value as netstring
356(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them). 492(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
357 493
358=back
359
360=cut 494=cut
361 495
362register_write_type netstring => sub { 496register_write_type netstring => sub {
363 my ($self, $string) = @_; 497 my ($self, $string) = @_;
364 498
365 sprintf "%d:%s,", (length $string), $string 499 sprintf "%d:%s,", (length $string), $string
366}; 500};
367 501
502=item packstring => $format, $data
503
504An octet string prefixed with an encoded length. The encoding C<$format>
505uses the same format as a Perl C<pack> format, but must specify a single
506integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
507optional C<!>, C<< < >> or C<< > >> modifier).
508
509=cut
510
511register_write_type packstring => sub {
512 my ($self, $format, $string) = @_;
513
514 pack "$format/a*", $string
515};
516
517=item json => $array_or_hashref
518
519Encodes the given hash or array reference into a JSON object. Unless you
520provide your own JSON object, this means it will be encoded to JSON text
521in UTF-8.
522
523JSON objects (and arrays) are self-delimiting, so you can write JSON at
524one end of a handle and read them at the other end without using any
525additional framing.
526
527The generated JSON text is guaranteed not to contain any newlines: While
528this module doesn't need delimiters after or between JSON texts to be
529able to read them, many other languages depend on that.
530
531A simple RPC protocol that interoperates easily with others is to send
532JSON arrays (or objects, although arrays are usually the better choice as
533they mimic how function argument passing works) and a newline after each
534JSON text:
535
536 $handle->push_write (json => ["method", "arg1", "arg2"]); # whatever
537 $handle->push_write ("\012");
538
539An AnyEvent::Handle receiver would simply use the C<json> read type and
540rely on the fact that the newline will be skipped as leading whitespace:
541
542 $handle->push_read (json => sub { my $array = $_[1]; ... });
543
544Other languages could read single lines terminated by a newline and pass
545this line into their JSON decoder of choice.
546
547=cut
548
549register_write_type json => sub {
550 my ($self, $ref) = @_;
551
552 require JSON;
553
554 $self->{json} ? $self->{json}->encode ($ref)
555 : JSON::encode_json ($ref)
556};
557
558=item storable => $reference
559
560Freezes the given reference using L<Storable> and writes it to the
561handle. Uses the C<nfreeze> format.
562
563=cut
564
565register_write_type storable => sub {
566 my ($self, $ref) = @_;
567
568 require Storable;
569
570 pack "w/a*", Storable::nfreeze ($ref)
571};
572
573=back
574
368=item AnyEvent::Handle::register_write_type type => $coderef->($self, @args) 575=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
369 576
370This function (not method) lets you add your own types to C<push_write>. 577This function (not method) lets you add your own types to C<push_write>.
371Whenever the given C<type> is used, C<push_write> will invoke the code 578Whenever the given C<type> is used, C<push_write> will invoke the code
372reference with the handle object and the remaining arguments. 579reference with the handle object and the remaining arguments.
373 580
392ways, the "simple" way, using only C<on_read> and the "complex" way, using 599ways, the "simple" way, using only C<on_read> and the "complex" way, using
393a queue. 600a queue.
394 601
395In the simple case, you just install an C<on_read> callback and whenever 602In the simple case, you just install an C<on_read> callback and whenever
396new data arrives, it will be called. You can then remove some data (if 603new data arrives, it will be called. You can then remove some data (if
397enough is there) from the read buffer (C<< $handle->rbuf >>) if you want 604enough is there) from the read buffer (C<< $handle->rbuf >>). Or you cna
398or not. 605leave the data there if you want to accumulate more (e.g. when only a
606partial message has been received so far).
399 607
400In the more complex case, you want to queue multiple callbacks. In this 608In the more complex case, you want to queue multiple callbacks. In this
401case, AnyEvent::Handle will call the first queued callback each time new 609case, AnyEvent::Handle will call the first queued callback each time new
402data arrives and removes it when it has done its job (see C<push_read>, 610data arrives (also the first time it is queued) and removes it when it has
403below). 611done its job (see C<push_read>, below).
404 612
405This way you can, for example, push three line-reads, followed by reading 613This way you can, for example, push three line-reads, followed by reading
406a chunk of data, and AnyEvent::Handle will execute them in order. 614a chunk of data, and AnyEvent::Handle will execute them in order.
407 615
408Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by 616Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
409the specified number of bytes which give an XML datagram. 617the specified number of bytes which give an XML datagram.
410 618
411 # in the default state, expect some header bytes 619 # in the default state, expect some header bytes
412 $handle->on_read (sub { 620 $handle->on_read (sub {
413 # some data is here, now queue the length-header-read (4 octets) 621 # some data is here, now queue the length-header-read (4 octets)
414 shift->unshift_read_chunk (4, sub { 622 shift->unshift_read (chunk => 4, sub {
415 # header arrived, decode 623 # header arrived, decode
416 my $len = unpack "N", $_[1]; 624 my $len = unpack "N", $_[1];
417 625
418 # now read the payload 626 # now read the payload
419 shift->unshift_read_chunk ($len, sub { 627 shift->unshift_read (chunk => $len, sub {
420 my $xml = $_[1]; 628 my $xml = $_[1];
421 # handle xml 629 # handle xml
422 }); 630 });
423 }); 631 });
424 }); 632 });
425 633
426Example 2: Implement a client for a protocol that replies either with 634Example 2: Implement a client for a protocol that replies either with "OK"
427"OK" and another line or "ERROR" for one request, and 64 bytes for the 635and another line or "ERROR" for the first request that is sent, and 64
428second request. Due tot he availability of a full queue, we can just 636bytes for the second request. Due to the availability of a queue, we can
429pipeline sending both requests and manipulate the queue as necessary in 637just pipeline sending both requests and manipulate the queue as necessary
430the callbacks: 638in the callbacks.
431 639
432 # request one 640When the first callback is called and sees an "OK" response, it will
641C<unshift> another line-read. This line-read will be queued I<before> the
64264-byte chunk callback.
643
644 # request one, returns either "OK + extra line" or "ERROR"
433 $handle->push_write ("request 1\015\012"); 645 $handle->push_write ("request 1\015\012");
434 646
435 # we expect "ERROR" or "OK" as response, so push a line read 647 # we expect "ERROR" or "OK" as response, so push a line read
436 $handle->push_read_line (sub { 648 $handle->push_read (line => sub {
437 # if we got an "OK", we have to _prepend_ another line, 649 # if we got an "OK", we have to _prepend_ another line,
438 # so it will be read before the second request reads its 64 bytes 650 # so it will be read before the second request reads its 64 bytes
439 # which are already in the queue when this callback is called 651 # which are already in the queue when this callback is called
440 # we don't do this in case we got an error 652 # we don't do this in case we got an error
441 if ($_[1] eq "OK") { 653 if ($_[1] eq "OK") {
442 $_[0]->unshift_read_line (sub { 654 $_[0]->unshift_read (line => sub {
443 my $response = $_[1]; 655 my $response = $_[1];
444 ... 656 ...
445 }); 657 });
446 } 658 }
447 }); 659 });
448 660
449 # request two 661 # request two, simply returns 64 octets
450 $handle->push_write ("request 2\015\012"); 662 $handle->push_write ("request 2\015\012");
451 663
452 # simply read 64 bytes, always 664 # simply read 64 bytes, always
453 $handle->push_read_chunk (64, sub { 665 $handle->push_read (chunk => 64, sub {
454 my $response = $_[1]; 666 my $response = $_[1];
455 ... 667 ...
456 }); 668 });
457 669
458=over 4 670=over 4
459 671
460=cut 672=cut
461 673
462sub _drain_rbuf { 674sub _drain_rbuf {
463 my ($self) = @_; 675 my ($self) = @_;
676
677 local $self->{_in_drain} = 1;
464 678
465 if ( 679 if (
466 defined $self->{rbuf_max} 680 defined $self->{rbuf_max}
467 && $self->{rbuf_max} < length $self->{rbuf} 681 && $self->{rbuf_max} < length $self->{rbuf}
468 ) { 682 ) {
469 $! = &Errno::ENOSPC; return $self->error; 683 return $self->_error (&Errno::ENOSPC, 1);
470 } 684 }
471 685
472 return if $self->{in_drain}; 686 while () {
473 local $self->{in_drain} = 1;
474
475 while (my $len = length $self->{rbuf}) {
476 no strict 'refs'; 687 no strict 'refs';
688
689 my $len = length $self->{rbuf};
690
477 if (my $cb = shift @{ $self->{queue} }) { 691 if (my $cb = shift @{ $self->{_queue} }) {
478 unless ($cb->($self)) { 692 unless ($cb->($self)) {
479 if ($self->{eof}) { 693 if ($self->{_eof}) {
480 # no progress can be made (not enough data and no data forthcoming) 694 # no progress can be made (not enough data and no data forthcoming)
481 $! = &Errno::EPIPE; return $self->error; 695 $self->_error (&Errno::EPIPE, 1), last;
482 } 696 }
483 697
484 unshift @{ $self->{queue} }, $cb; 698 unshift @{ $self->{_queue} }, $cb;
485 return; 699 last;
486 } 700 }
487 } elsif ($self->{on_read}) { 701 } elsif ($self->{on_read}) {
702 last unless $len;
703
488 $self->{on_read}($self); 704 $self->{on_read}($self);
489 705
490 if ( 706 if (
491 $self->{eof} # if no further data will arrive
492 && $len == length $self->{rbuf} # and no data has been consumed 707 $len == length $self->{rbuf} # if no data has been consumed
493 && !@{ $self->{queue} } # and the queue is still empty 708 && !@{ $self->{_queue} } # and the queue is still empty
494 && $self->{on_read} # and we still want to read data 709 && $self->{on_read} # but we still have on_read
495 ) { 710 ) {
711 # no further data will arrive
496 # then no progress can be made 712 # so no progress can be made
497 $! = &Errno::EPIPE; return $self->error; 713 $self->_error (&Errno::EPIPE, 1), last
714 if $self->{_eof};
715
716 last; # more data might arrive
498 } 717 }
499 } else { 718 } else {
500 # read side becomes idle 719 # read side becomes idle
501 delete $self->{rw}; 720 delete $self->{_rw};
502 return; 721 last;
503 } 722 }
504 } 723 }
505 724
506 if ($self->{eof}) {
507 $self->_shutdown;
508 $self->{on_eof}($self) 725 $self->{on_eof}($self)
509 if $self->{on_eof}; 726 if $self->{_eof} && $self->{on_eof};
727
728 # may need to restart read watcher
729 unless ($self->{_rw}) {
730 $self->start_read
731 if $self->{on_read} || @{ $self->{_queue} };
510 } 732 }
511} 733}
512 734
513=item $handle->on_read ($cb) 735=item $handle->on_read ($cb)
514 736
520 742
521sub on_read { 743sub on_read {
522 my ($self, $cb) = @_; 744 my ($self, $cb) = @_;
523 745
524 $self->{on_read} = $cb; 746 $self->{on_read} = $cb;
747 $self->_drain_rbuf if $cb && !$self->{_in_drain};
525} 748}
526 749
527=item $handle->rbuf 750=item $handle->rbuf
528 751
529Returns the read buffer (as a modifiable lvalue). 752Returns the read buffer (as a modifiable lvalue).
577 800
578 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 801 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
579 ->($self, $cb, @_); 802 ->($self, $cb, @_);
580 } 803 }
581 804
582 push @{ $self->{queue} }, $cb; 805 push @{ $self->{_queue} }, $cb;
583 $self->_drain_rbuf; 806 $self->_drain_rbuf unless $self->{_in_drain};
584} 807}
585 808
586sub unshift_read { 809sub unshift_read {
587 my $self = shift; 810 my $self = shift;
588 my $cb = pop; 811 my $cb = pop;
593 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read") 816 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
594 ->($self, $cb, @_); 817 ->($self, $cb, @_);
595 } 818 }
596 819
597 820
598 unshift @{ $self->{queue} }, $cb; 821 unshift @{ $self->{_queue} }, $cb;
599 $self->_drain_rbuf; 822 $self->_drain_rbuf unless $self->{_in_drain};
600} 823}
601 824
602=item $handle->push_read (type => @args, $cb) 825=item $handle->push_read (type => @args, $cb)
603 826
604=item $handle->unshift_read (type => @args, $cb) 827=item $handle->unshift_read (type => @args, $cb)
610Predefined types are (if you have ideas for additional types, feel free to 833Predefined types are (if you have ideas for additional types, feel free to
611drop by and tell us): 834drop by and tell us):
612 835
613=over 4 836=over 4
614 837
615=item chunk => $octets, $cb->($self, $data) 838=item chunk => $octets, $cb->($handle, $data)
616 839
617Invoke the callback only once C<$octets> bytes have been read. Pass the 840Invoke the callback only once C<$octets> bytes have been read. Pass the
618data read to the callback. The callback will never be called with less 841data read to the callback. The callback will never be called with less
619data. 842data.
620 843
643 866
644sub unshift_read_chunk { 867sub unshift_read_chunk {
645 $_[0]->unshift_read (chunk => $_[1], $_[2]); 868 $_[0]->unshift_read (chunk => $_[1], $_[2]);
646} 869}
647 870
648=item line => [$eol, ]$cb->($self, $line, $eol) 871=item line => [$eol, ]$cb->($handle, $line, $eol)
649 872
650The callback will be called only once a full line (including the end of 873The callback will be called only once a full line (including the end of
651line marker, C<$eol>) has been read. This line (excluding the end of line 874line marker, C<$eol>) has been read. This line (excluding the end of line
652marker) will be passed to the callback as second argument (C<$line>), and 875marker) will be passed to the callback as second argument (C<$line>), and
653the end of line marker as the third argument (C<$eol>). 876the end of line marker as the third argument (C<$eol>).
690sub unshift_read_line { 913sub unshift_read_line {
691 my $self = shift; 914 my $self = shift;
692 $self->unshift_read (line => @_); 915 $self->unshift_read (line => @_);
693} 916}
694 917
918=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
919
920Makes a regex match against the regex object C<$accept> and returns
921everything up to and including the match.
922
923Example: read a single line terminated by '\n'.
924
925 $handle->push_read (regex => qr<\n>, sub { ... });
926
927If C<$reject> is given and not undef, then it determines when the data is
928to be rejected: it is matched against the data when the C<$accept> regex
929does not match and generates an C<EBADMSG> error when it matches. This is
930useful to quickly reject wrong data (to avoid waiting for a timeout or a
931receive buffer overflow).
932
933Example: expect a single decimal number followed by whitespace, reject
934anything else (not the use of an anchor).
935
936 $handle->push_read (regex => qr<^[0-9]+\s>, qr<[^0-9]>, sub { ... });
937
938If C<$skip> is given and not C<undef>, then it will be matched against
939the receive buffer when neither C<$accept> nor C<$reject> match,
940and everything preceding and including the match will be accepted
941unconditionally. This is useful to skip large amounts of data that you
942know cannot be matched, so that the C<$accept> or C<$reject> regex do not
943have to start matching from the beginning. This is purely an optimisation
944and is usually worth only when you expect more than a few kilobytes.
945
946Example: expect a http header, which ends at C<\015\012\015\012>. Since we
947expect the header to be very large (it isn't in practise, but...), we use
948a skip regex to skip initial portions. The skip regex is tricky in that
949it only accepts something not ending in either \015 or \012, as these are
950required for the accept regex.
951
952 $handle->push_read (regex =>
953 qr<\015\012\015\012>,
954 undef, # no reject
955 qr<^.*[^\015\012]>,
956 sub { ... });
957
958=cut
959
960register_read_type regex => sub {
961 my ($self, $cb, $accept, $reject, $skip) = @_;
962
963 my $data;
964 my $rbuf = \$self->{rbuf};
965
966 sub {
967 # accept
968 if ($$rbuf =~ $accept) {
969 $data .= substr $$rbuf, 0, $+[0], "";
970 $cb->($self, $data);
971 return 1;
972 }
973
974 # reject
975 if ($reject && $$rbuf =~ $reject) {
976 $self->_error (&Errno::EBADMSG);
977 }
978
979 # skip
980 if ($skip && $$rbuf =~ $skip) {
981 $data .= substr $$rbuf, 0, $+[0], "";
982 }
983
984 ()
985 }
986};
987
695=item netstring => $cb->($string) 988=item netstring => $cb->($handle, $string)
696 989
697A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement). 990A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
698 991
699Throws an error with C<$!> set to EBADMSG on format violations. 992Throws an error with C<$!> set to EBADMSG on format violations.
700 993
704 my ($self, $cb) = @_; 997 my ($self, $cb) = @_;
705 998
706 sub { 999 sub {
707 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) { 1000 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
708 if ($_[0]{rbuf} =~ /[^0-9]/) { 1001 if ($_[0]{rbuf} =~ /[^0-9]/) {
709 $! = &Errno::EBADMSG; 1002 $self->_error (&Errno::EBADMSG);
710 $self->error;
711 } 1003 }
712 return; 1004 return;
713 } 1005 }
714 1006
715 my $len = $1; 1007 my $len = $1;
718 my $string = $_[1]; 1010 my $string = $_[1];
719 $_[0]->unshift_read (chunk => 1, sub { 1011 $_[0]->unshift_read (chunk => 1, sub {
720 if ($_[1] eq ",") { 1012 if ($_[1] eq ",") {
721 $cb->($_[0], $string); 1013 $cb->($_[0], $string);
722 } else { 1014 } else {
723 $! = &Errno::EBADMSG; 1015 $self->_error (&Errno::EBADMSG);
724 $self->error;
725 } 1016 }
726 }); 1017 });
727 }); 1018 });
728 1019
729 1 1020 1
730 } 1021 }
731}; 1022};
732 1023
1024=item packstring => $format, $cb->($handle, $string)
1025
1026An octet string prefixed with an encoded length. The encoding C<$format>
1027uses the same format as a Perl C<pack> format, but must specify a single
1028integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
1029optional C<!>, C<< < >> or C<< > >> modifier).
1030
1031DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
1032
1033Example: read a block of data prefixed by its length in BER-encoded
1034format (very efficient).
1035
1036 $handle->push_read (packstring => "w", sub {
1037 my ($handle, $data) = @_;
1038 });
1039
1040=cut
1041
1042register_read_type packstring => sub {
1043 my ($self, $cb, $format) = @_;
1044
1045 sub {
1046 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1047 defined (my $len = eval { unpack $format, $_[0]->{rbuf} })
1048 or return;
1049
1050 # remove prefix
1051 substr $_[0]->{rbuf}, 0, (length pack $format, $len), "";
1052
1053 # read rest
1054 $_[0]->unshift_read (chunk => $len, $cb);
1055
1056 1
1057 }
1058};
1059
1060=item json => $cb->($handle, $hash_or_arrayref)
1061
1062Reads a JSON object or array, decodes it and passes it to the callback.
1063
1064If a C<json> object was passed to the constructor, then that will be used
1065for the final decode, otherwise it will create a JSON coder expecting UTF-8.
1066
1067This read type uses the incremental parser available with JSON version
10682.09 (and JSON::XS version 2.2) and above. You have to provide a
1069dependency on your own: this module will load the JSON module, but
1070AnyEvent does not depend on it itself.
1071
1072Since JSON texts are fully self-delimiting, the C<json> read and write
1073types are an ideal simple RPC protocol: just exchange JSON datagrams. See
1074the C<json> write type description, above, for an actual example.
1075
1076=cut
1077
1078register_read_type json => sub {
1079 my ($self, $cb) = @_;
1080
1081 require JSON;
1082
1083 my $data;
1084 my $rbuf = \$self->{rbuf};
1085
1086 my $json = $self->{json} ||= JSON->new->utf8;
1087
1088 sub {
1089 my $ref = $json->incr_parse ($self->{rbuf});
1090
1091 if ($ref) {
1092 $self->{rbuf} = $json->incr_text;
1093 $json->incr_text = "";
1094 $cb->($self, $ref);
1095
1096 1
1097 } else {
1098 $self->{rbuf} = "";
1099 ()
1100 }
1101 }
1102};
1103
1104=item storable => $cb->($handle, $ref)
1105
1106Deserialises a L<Storable> frozen representation as written by the
1107C<storable> write type (BER-encoded length prefix followed by nfreeze'd
1108data).
1109
1110Raises C<EBADMSG> error if the data could not be decoded.
1111
1112=cut
1113
1114register_read_type storable => sub {
1115 my ($self, $cb) = @_;
1116
1117 require Storable;
1118
1119 sub {
1120 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1121 defined (my $len = eval { unpack "w", $_[0]->{rbuf} })
1122 or return;
1123
1124 # remove prefix
1125 substr $_[0]->{rbuf}, 0, (length pack "w", $len), "";
1126
1127 # read rest
1128 $_[0]->unshift_read (chunk => $len, sub {
1129 if (my $ref = eval { Storable::thaw ($_[1]) }) {
1130 $cb->($_[0], $ref);
1131 } else {
1132 $self->_error (&Errno::EBADMSG);
1133 }
1134 });
1135 }
1136};
1137
733=back 1138=back
734 1139
735=item AnyEvent::Handle::register_read_type type => $coderef->($self, $cb, @args) 1140=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
736 1141
737This function (not method) lets you add your own types to C<push_read>. 1142This function (not method) lets you add your own types to C<push_read>.
738 1143
739Whenever the given C<type> is used, C<push_read> will invoke the code 1144Whenever the given C<type> is used, C<push_read> will invoke the code
740reference with the handle object, the callback and the remaining 1145reference with the handle object, the callback and the remaining
742 1147
743The code reference is supposed to return a callback (usually a closure) 1148The code reference is supposed to return a callback (usually a closure)
744that works as a plain read callback (see C<< ->push_read ($cb) >>). 1149that works as a plain read callback (see C<< ->push_read ($cb) >>).
745 1150
746It should invoke the passed callback when it is done reading (remember to 1151It should invoke the passed callback when it is done reading (remember to
747pass C<$self> as first argument as all other callbacks do that). 1152pass C<$handle> as first argument as all other callbacks do that).
748 1153
749Note that this is a function, and all types registered this way will be 1154Note that this is a function, and all types registered this way will be
750global, so try to use unique names. 1155global, so try to use unique names.
751 1156
752For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>, 1157For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>,
755=item $handle->stop_read 1160=item $handle->stop_read
756 1161
757=item $handle->start_read 1162=item $handle->start_read
758 1163
759In rare cases you actually do not want to read anything from the 1164In rare cases you actually do not want to read anything from the
760socket. In this case you can call C<stop_read>. Neither C<on_read> no 1165socket. In this case you can call C<stop_read>. Neither C<on_read> nor
761any queued callbacks will be executed then. To start reading again, call 1166any queued callbacks will be executed then. To start reading again, call
762C<start_read>. 1167C<start_read>.
763 1168
1169Note that AnyEvent::Handle will automatically C<start_read> for you when
1170you change the C<on_read> callback or push/unshift a read callback, and it
1171will automatically C<stop_read> for you when neither C<on_read> is set nor
1172there are any read requests in the queue.
1173
764=cut 1174=cut
765 1175
766sub stop_read { 1176sub stop_read {
767 my ($self) = @_; 1177 my ($self) = @_;
768 1178
769 delete $self->{rw}; 1179 delete $self->{_rw};
770} 1180}
771 1181
772sub start_read { 1182sub start_read {
773 my ($self) = @_; 1183 my ($self) = @_;
774 1184
775 unless ($self->{rw} || $self->{eof}) { 1185 unless ($self->{_rw} || $self->{_eof}) {
776 Scalar::Util::weaken $self; 1186 Scalar::Util::weaken $self;
777 1187
778 $self->{rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1188 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
779 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1189 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
780 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1190 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
781 1191
782 if ($len > 0) { 1192 if ($len > 0) {
1193 $self->{_activity} = AnyEvent->now;
1194
783 $self->{filter_r} 1195 $self->{filter_r}
784 ? $self->{filter_r}->($self, $rbuf) 1196 ? $self->{filter_r}($self, $rbuf)
785 : $self->_drain_rbuf; 1197 : $self->{_in_drain} || $self->_drain_rbuf;
786 1198
787 } elsif (defined $len) { 1199 } elsif (defined $len) {
788 delete $self->{rw}; 1200 delete $self->{_rw};
789 $self->{eof} = 1; 1201 $self->{_eof} = 1;
790 $self->_drain_rbuf; 1202 $self->_drain_rbuf unless $self->{_in_drain};
791 1203
792 } elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) { 1204 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
793 return $self->error; 1205 return $self->_error ($!, 1);
794 } 1206 }
795 }); 1207 });
796 } 1208 }
797} 1209}
798 1210
799sub _dotls { 1211sub _dotls {
800 my ($self) = @_; 1212 my ($self) = @_;
801 1213
1214 my $buf;
1215
802 if (length $self->{tls_wbuf}) { 1216 if (length $self->{_tls_wbuf}) {
803 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{tls_wbuf})) > 0) { 1217 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
804 substr $self->{tls_wbuf}, 0, $len, ""; 1218 substr $self->{_tls_wbuf}, 0, $len, "";
805 } 1219 }
806 } 1220 }
807 1221
808 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{tls_wbio}))) { 1222 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
809 $self->{wbuf} .= $buf; 1223 $self->{wbuf} .= $buf;
810 $self->_drain_wbuf; 1224 $self->_drain_wbuf;
811 } 1225 }
812 1226
813 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1227 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1228 if (length $buf) {
814 $self->{rbuf} .= $buf; 1229 $self->{rbuf} .= $buf;
815 $self->_drain_rbuf; 1230 $self->_drain_rbuf unless $self->{_in_drain};
1231 } else {
1232 # let's treat SSL-eof as we treat normal EOF
1233 $self->{_eof} = 1;
1234 $self->_shutdown;
1235 return;
1236 }
816 } 1237 }
817 1238
818 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1239 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
819 1240
820 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1241 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
821 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1242 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
822 $self->error; 1243 return $self->_error ($!, 1);
823 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1244 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
824 $! = &Errno::EIO; 1245 return $self->_error (&Errno::EIO, 1);
825 $self->error;
826 } 1246 }
827 1247
828 # all others are fine for our purposes 1248 # all others are fine for our purposes
829 } 1249 }
830} 1250}
839C<"connect">, C<"accept"> or an existing Net::SSLeay object). 1259C<"connect">, C<"accept"> or an existing Net::SSLeay object).
840 1260
841The second argument is the optional C<Net::SSLeay::CTX> object that is 1261The second argument is the optional C<Net::SSLeay::CTX> object that is
842used when AnyEvent::Handle has to create its own TLS connection object. 1262used when AnyEvent::Handle has to create its own TLS connection object.
843 1263
844=cut 1264The TLS connection object will end up in C<< $handle->{tls} >> after this
1265call and can be used or changed to your liking. Note that the handshake
1266might have already started when this function returns.
845 1267
846# TODO: maybe document... 1268=cut
1269
847sub starttls { 1270sub starttls {
848 my ($self, $ssl, $ctx) = @_; 1271 my ($self, $ssl, $ctx) = @_;
849 1272
850 $self->stoptls; 1273 $self->stoptls;
851 1274
866 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html 1289 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
867 Net::SSLeay::CTX_set_mode ($self->{tls}, 1290 Net::SSLeay::CTX_set_mode ($self->{tls},
868 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) 1291 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
869 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2)); 1292 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
870 1293
871 $self->{tls_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1294 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
872 $self->{tls_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1295 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
873 1296
874 Net::SSLeay::set_bio ($ssl, $self->{tls_rbio}, $self->{tls_wbio}); 1297 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio});
875 1298
876 $self->{filter_w} = sub { 1299 $self->{filter_w} = sub {
877 $_[0]{tls_wbuf} .= ${$_[1]}; 1300 $_[0]{_tls_wbuf} .= ${$_[1]};
878 &_dotls; 1301 &_dotls;
879 }; 1302 };
880 $self->{filter_r} = sub { 1303 $self->{filter_r} = sub {
881 Net::SSLeay::BIO_write ($_[0]{tls_rbio}, ${$_[1]}); 1304 Net::SSLeay::BIO_write ($_[0]{_rbio}, ${$_[1]});
882 &_dotls; 1305 &_dotls;
883 }; 1306 };
884} 1307}
885 1308
886=item $handle->stoptls 1309=item $handle->stoptls
892 1315
893sub stoptls { 1316sub stoptls {
894 my ($self) = @_; 1317 my ($self) = @_;
895 1318
896 Net::SSLeay::free (delete $self->{tls}) if $self->{tls}; 1319 Net::SSLeay::free (delete $self->{tls}) if $self->{tls};
1320
897 delete $self->{tls_rbio}; 1321 delete $self->{_rbio};
898 delete $self->{tls_wbio}; 1322 delete $self->{_wbio};
899 delete $self->{tls_wbuf}; 1323 delete $self->{_tls_wbuf};
900 delete $self->{filter_r}; 1324 delete $self->{filter_r};
901 delete $self->{filter_w}; 1325 delete $self->{filter_w};
902} 1326}
903 1327
904sub DESTROY { 1328sub DESTROY {
905 my $self = shift; 1329 my $self = shift;
906 1330
907 $self->stoptls; 1331 $self->stoptls;
1332
1333 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1334
1335 if ($linger && length $self->{wbuf}) {
1336 my $fh = delete $self->{fh};
1337 my $wbuf = delete $self->{wbuf};
1338
1339 my @linger;
1340
1341 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {
1342 my $len = syswrite $fh, $wbuf, length $wbuf;
1343
1344 if ($len > 0) {
1345 substr $wbuf, 0, $len, "";
1346 } else {
1347 @linger = (); # end
1348 }
1349 });
1350 push @linger, AnyEvent->timer (after => $linger, cb => sub {
1351 @linger = ();
1352 });
1353 }
908} 1354}
909 1355
910=item AnyEvent::Handle::TLS_CTX 1356=item AnyEvent::Handle::TLS_CTX
911 1357
912This function creates and returns the Net::SSLeay::CTX object used by 1358This function creates and returns the Net::SSLeay::CTX object used by
942 } 1388 }
943} 1389}
944 1390
945=back 1391=back
946 1392
1393=head1 SUBCLASSING AnyEvent::Handle
1394
1395In many cases, you might want to subclass AnyEvent::Handle.
1396
1397To make this easier, a given version of AnyEvent::Handle uses these
1398conventions:
1399
1400=over 4
1401
1402=item * all constructor arguments become object members.
1403
1404At least initially, when you pass a C<tls>-argument to the constructor it
1405will end up in C<< $handle->{tls} >>. Those members might be changes or
1406mutated later on (for example C<tls> will hold the TLS connection object).
1407
1408=item * other object member names are prefixed with an C<_>.
1409
1410All object members not explicitly documented (internal use) are prefixed
1411with an underscore character, so the remaining non-C<_>-namespace is free
1412for use for subclasses.
1413
1414=item * all members not documented here and not prefixed with an underscore
1415are free to use in subclasses.
1416
1417Of course, new versions of AnyEvent::Handle may introduce more "public"
1418member variables, but thats just life, at least it is documented.
1419
1420=back
1421
947=head1 AUTHOR 1422=head1 AUTHOR
948 1423
949Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>. 1424Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
950 1425
951=cut 1426=cut

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines