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.43 by root, Wed May 28 23:57:38 2008 UTC vs.
Revision 1.87 by root, Thu Aug 21 20:52:39 2008 UTC

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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines