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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines